package net.wlm.job;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
public class VehicleJob {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 2. 输入表-input:映射到kafka消息队列
        createInputTable(tabEnv);
        // 3.数据处理
        handle(tabEnv);
        // 3. 输出表-output:映射到clickhouse
        createOutputTable(tabEnv);
        // 4. 保存数据
        saveToClickHouse(tabEnv);
    }
    private static void saveToClickHouse(TableEnvironment tabEnv) {
        // 临时表数据插入CK表
        tabEnv.executeSql("insert into  vehicle_kafka_sink(" +
                "vin, `timestamp`, longitude, latitude, province,speed, mileage,batteryLevel, batteryTemp, motorTemp, " +
                "chargingStatus, energyConsumption)\n" +
                "select\n" +
                "vin, `timestamp`, longitude, latitude, province, speed, mileage, batteryLevel, batteryTemp, motorTemp," +
                " chargingStatus, energyConsumption\n" +
                "from vehicle_log_tmp");
    }
    private static void handle(TableEnvironment tabEnv) {
        // 拿到kafka输入表数据，存放临时表
        Table vehicle_log_kafka_source = tabEnv.sqlQuery("select * from vehicle_log_kafka_source");
        tabEnv.createTemporaryView("vehicle_log_tmp", vehicle_log_kafka_source);
    }
    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE vehicle_kafka_sink (\n" +
                "       vin               varchar(64) comment '车辆唯一标识',\n" +
                "       `timestamp`       bigint comment '数据时间戳',\n" +
                "       longitude         float comment '经度',\n" +
                "       latitude          float comment '维度',\n" +
                "       province          varchar(8) comment '省',\n" +
                "       speed             float comment '车速(km/h)',\n" +
                "       mileage           float comment '总里程(km)',\n" +
                "       batteryLevel      float comment '电池电量(%)',\n" +
                "       batteryTemp       float comment '电池温度(℃)',\n" +
                "       motorTemp         float comment '电机温度(℃)',\n" +
                "       chargingStatus    varchar(8) comment '充电状态：charging(充电)、discharging(放电)',\n" +
                "       energyConsumption float comment '瞬时能耗'\n" +
                ") WITH (\n" +
                "      'connector' = 'clickhouse',\n" +
                "      'url' = 'jdbc:clickhouse://node103:8123/vehicle_real',\n" +
                "      'table' = 'vehicle_log',\n" +
                "      'username' = 'default',\n" +
                "      'password' = '',\n" +
                "      'format' = 'json'\n" +
                "      )");
    }
    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE vehicle_log_kafka_source (\n" +
                "       vin               varchar(64) comment '车辆唯一标识' ,\n" +
                "       `timestamp`                bigint comment '数据时间戳' ,\n" +
                "       longitude         float comment '经度',\n" +
                "       latitude          float comment '维度',\n" +
                "       province          varchar(8) comment '省',\n" +
                "       speed             float comment '车速(km/h)',\n" +
                "       mileage           float comment '总里程(km)',\n" +
                "       batteryLevel      float comment '电池电量(%)',\n" +
                "       batteryTemp       float comment '电池温度(℃)',\n" +
                "       motorTemp         float comment '电机温度(℃)',\n" +
                "       chargingStatus    varchar(8) comment '充电状态：charging(充电)、discharging(放电)',\n" +
                "       energyConsumption float comment '瞬时能耗'\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'vehicle-log',\n" +
                " 'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                " 'properties.group.id' = 'testGroup',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json',\n" +
                " 'json.fail-on-missing-field' = 'false',\n" +
                " 'json.ignore-parse-errors' = 'true'\n" +
                ")");
    }
    private static TableEnvironment getTableEnv() {
        // a-环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // b-属性配置设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // c-返回对象
        return tabEnv;
    }
}